﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.Samples.BizTalk.Adapter.Common;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.BizTalk.Component.Interop;
using System.IO;
using Microsoft.BizTalk.Streaming;
using ZMQ;
using System.Threading;

namespace com.fluister.BizTalk0MQAdapter.BizTalk0MQSendAdapter
{
    class BizTalk0MQSendAdapterEndpoint : AsyncTransmitterEndpoint
    {
        private IBaseMessage    solicitMsg                    = null;
        private IBTTransportProxy    transportProxy                = null;
        private AsyncTransmitter    asyncTransmitter            = null;
        private string propertyNamespace;
        private static int IO_BUFFER_SIZE = 4096;

        public BizTalk0MQSendAdapterEndpoint(AsyncTransmitter asyncTransmitter)
            : base(asyncTransmitter)
        {
            this.asyncTransmitter = asyncTransmitter;
            this.transportProxy = asyncTransmitter.TransportProxy;
        }

        public override void Open(EndpointParameters endpointParameters, IPropertyBag handlerPropertyBag, string propertyNamespace)
        {
            this.propertyNamespace = propertyNamespace;
        }


        /// <summary>
        /// Implementation for AsyncTransmitterEndpoint::ProcessMessage
        /// Transmit the message and optionally return the response message
        /// </summary>
        public override IBaseMessage ProcessMessage(IBaseMessage message)
        {
            this.solicitMsg = message;

            BizTalk0MQAdapterProperties props = new BizTalk0MQAdapterProperties(message, this.propertyNamespace);
            IBaseMessage responseMsg = null;
            
            String response = Send0MQRequest(this.solicitMsg, props);
            responseMsg = BuildResponseMessage(response);

            return responseMsg;
        }

        private IBaseMessage BuildResponseMessage(String zmqResponse)
        {
            IBaseMessage btsResponse = null;

            // Create stream from string response
            MemoryStream outStream = new MemoryStream();
            byte[] buff = new byte[zmqResponse.Length * sizeof(char)];
            System.Buffer.BlockCopy(zmqResponse.ToCharArray(), 0, buff, 0, buff.Length);                                   
            outStream.Write(buff, 0, buff.Length);
            
            // Construct iBaseMessage
            IBaseMessageFactory mf = transportProxy.GetMessageFactory();
            btsResponse = mf.CreateMessage();
            IBaseMessagePart body = mf.CreateMessagePart();
            body.Data = outStream;
            btsResponse.AddPart("Body", body, true);            

            return btsResponse;
        }

        private String Send0MQRequest(IBaseMessage msg, BizTalk0MQAdapterProperties config)
        {
            string response = null;
            #region SSO Usage Example
            // If we needed to use SSO to lookup the remote system credentials
            /*
            string ssoAffiliateApplication = props.AffiliateApplication;
            if (ssoAffiliateApplication.Length > 0)
            {
                SSOResult ssoResult = new SSOResult(msg, affiliateApplication);
                string userName  = ssoResult.UserName;
                string password  = ssoResult.Result[0];
                string etcEtcEtc = ssoResult.Result[1]; // (you can have additional metadata associated with the login in SSO) 
                    
                // use these credentials to login to the remote system
                // ideally zero off the password memory once we are done
            }                    
            */
            #endregion // SSO Usage Example
            using (var context = new Context(1))
            {
              
                SocketType type = SocketType.PUB;
                if (config.Method == "REQ_REP")
                    type = SocketType.REQ;
                using (Socket publisher = context.Socket(type))
                {
                    
                    publisher.Bind(config.Uri);
                    
                    IBaseMessagePart bodyPart = msg.BodyPart;
                    MemoryStream outs = new MemoryStream();
                    
                    int bytesRead = 0;
                    byte[] buff = new byte[IO_BUFFER_SIZE];
                    while ((bytesRead = bodyPart.Data.Read(buff, 0, IO_BUFFER_SIZE)) > 0)
                        outs.Write(buff, 0, bytesRead);
                    if (config.Method == "REQ_REP")
                    {
                        /// TODO: Add subject property
                        // publisher.SendMore(subject, Encoding.Unicode);
                        SendStatus status = publisher.Send(Encoding.UTF8.GetString(outs.ToArray()), Encoding.Unicode);
                        
                        if (status == SendStatus.Sent)
                        {
                            // Wait for reply or timeout
                            response = publisher.Recv(Encoding.Unicode, config.Timeout);
                        }
                    }
                    else // PUB SUB(Reliable only)
                    {
                        /// TODO: Add subject property
                        // publisher.SendMore(subject, Encoding.Unicode);
                        SendStatus status =  publisher.Send(Encoding.UTF8.GetString(outs.ToArray()), Encoding.Unicode);
                        switch (status)
                        {
                            case SendStatus.Sent:
                                response = "Sent";
                                break;
                            case SendStatus.Interrupted:
                                response = "error";
                                break;
                        }
                    }
                }
            }
            return response;
        }
    }
}
